package com.netflix.hystrix.examples.myDemo;

import rx.Observable;
import rx.Observer;
import rx.Scheduler;
import rx.Subscription;
import rx.observables.SyncOnSubscribe;

/**
 * Description: 练习RXJAVA的demo
 *
 * @author Zhao WJ
 * @version 1.0
 * @date 2021/8/13 10:27
 */
public class RxjavaExample {

    /**
     * 功能描述:  Observable
     * @param args
     * @author Zhao WJ
     * @date 2021/8/13 10:29
     */
    public static void main(String[] args) {
        //被观察者 <返回值类型,观察者的泛型(onNext方法的入参类型)>
        //Observable调用subscribe方法订阅Observer后(被观察者订阅观察者), Observable会回调Observer的方法,
        //SyncOnSubscribe -> create方法接收的多种类型之一
        Observable<Integer> observable = Observable.create(new SyncOnSubscribe<String, Integer>() {

            @Override
            protected String generateState() {
                return "myState";
            }

            @Override
            protected String next(String state, Observer<? super Integer> observer) {
                observer.onNext(1);
                observer.onCompleted();
                return "next";
            }
        });

        //观察者
        //onError / onCompleted方法仅有一个方法会被调用并且只会调用一次
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Observer is completed.");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Observer is error! ");
            }

            @Override
            public void onNext(Integer o) {
                System.out.println("Observer next");
            }
        };

        //订阅
        //Subscription 可以取消订阅关系
        Subscription subscribe = observable.subscribe(observer);

        System.out.println(subscribe.isUnsubscribed());
        //取消订阅
        subscribe.unsubscribe();
        System.out.println(subscribe.isUnsubscribed());


    }
}
